/*
 * @(#)MultiThreadedCalculator.java    Jul 19, 2010
 *
 * Copyright 2009 Nyber Shanghai, Inc. All rights reserved.
 */

package com.lab.util.concurrent;

import java.util.ArrayList;
import java.util.List;
import java.util.concurrent.CountDownLatch;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;

/**
 * Say we have 100 numbers and we are break them into batches of size 3,
 * {1, 2, 3}, {4, 5, 6} .... {97, 98, 99}, {100}
 * now compute the sums, 6, 15, ..., 294, 100
 * There are 34 numbers. Now we should break this array again into batches of size 3, right?
 *
 * @author nickevin@gmail.com
 * @version Jul 19, 2010 9:31:01 AM
 */

public class MultiThreadedCalculator {

	private int numThreads = 0; // 0 means no threading  
	private int partSize = 100;
	private SingleThreadCalculator singleThreadCalculator = new SingleThreadCalculator();

	public MultiThreadedCalculator() {}

	public MultiThreadedCalculator(int numThreads, int partSize) {
		if (numThreads < 0 || partSize < 0) {
			throw new IllegalArgumentException(
					"The argument(s) can't be negative.");
		}

		this.numThreads = numThreads;
		this.partSize = partSize;
	}

	public long calc(long[] values) {
		if (numThreads == 0) {
			return singleThreadCalculator.calc(values, 0, values.length);
		}

		if (values == null || values.length == 0) {
			return 0;
		}

		// compute how many parts we can divide  
		int len = (values.length + partSize - 1) / partSize;

		long[] sums = new long[len]; // partial results  

		CountDownLatch doneSignal = new CountDownLatch(len);

		ExecutorService executor = Executors.newFixedThreadPool(numThreads);
		List<RunnableCalculator> calcList = new ArrayList<RunnableCalculator>(
				len);
		for (int i = 0; i < len; i++) {
			int start = i * partSize;
			int end = i == len - 1 ? values.length : start + partSize;

			RunnableCalculator rc = new RunnableCalculator(
					singleThreadCalculator, values, start, end, doneSignal);
			calcList.add(rc);
			executor.execute(rc);
		}

		try {
			doneSignal.await();
			for (int i = 0; i < len; i++) {
				sums[i] = calcList.get(i).getResult();
			}

			if (sums.length <= partSize) {
				return singleThreadCalculator.calc(sums, 0, sums.length);
			} else {
				return calc(sums);
			}
		} catch (InterruptedException ie) {
			throw new RuntimeException("got interrupted", ie);
		}
	}
}